package net.lmc.realtime.jtp.dim.job;

import net.lmc.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.lmc.realtime.jtp.dim.function.DimDataFilterFunction;
import net.lmc.realtime.jtp.dim.sink.DimHbaseSinkFunction;
import net.lmc.realtime.jtp.dim.function.HbaseDimMapFunction;
import net.lmc.realtime.jtp.dim.sink.DimHbaseSinkFunction1;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

public class RealtimeSyncDataDimJob {
    public static void main(String[] args) throws Exception {
        //环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L);

        //数据源
        DataStream<String> stream = KafkaUtil.consumerKafka(env, "topic_db");

        //数据转换
        DataStream<String> dimStream = handle(stream);
//dimStream.print();
        //数据接收
        dimStream.addSink(new DimHbaseSinkFunction1());

        //触发执行
        env.execute("RealtimeSyncDataDimJob");
    }

    private static DataStream<String> handle(DataStream<String> stream) {
        //维度表和主键映射
        HashMap<String, String> dimMap = new HashMap<String, String>() {{
            put("base_dic", "dic_code"); put("base_province", "id"); put("base_region", "id");
            put("base_category1", "id"); put("base_category2", "id"); put("base_category3", "id");
            put("sku_info", "id"); put("spu_info", "id"); put("user_info", "id");
            put("activity_info", "id"); put("activity_rule", "id"); put("coupon_info", "id");
            put("base_trademark","id");
        }};

        //维度表名称
        Set<String> dimSet = new HashSet<>(dimMap.keySet());

        //过滤维度表数据
        SingleOutputStreamOperator<String> syncStream = stream.filter(new DimDataFilterFunction(dimSet));
        //syncStream.print();
        //添加字段值
        SingleOutputStreamOperator<String> hbaseStream = syncStream.map(new HbaseDimMapFunction(dimMap));
        hbaseStream.print();
        return hbaseStream;
    }
}
